package com.darcytech.debezium.common.network;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.LocalDateTime;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;


public class AddressManager {

    private static final Logger LOGGER = LoggerFactory.getLogger(AddressManager.class);

    private final Set<Address> addresses;

    private final Long failureRetryTime;

    private final AtomicInteger count = new AtomicInteger();

    private final Thread addressMaintenanceThread;


    protected AddressManager(List<String> addressesList, long failureRetryTime) {
        addresses = new HashSet<>();
        this.failureRetryTime = failureRetryTime;
        for (String address : addressesList) {
            addresses.add(this.new Address(address, Boolean.TRUE, null));
        }
        addressMaintenanceThread = new Thread(this.new AddressMaintenance());
        addressMaintenanceThread.start();
    }

    public String getAvailableAddress() {
        if (this.addresses == null || this.addresses.size() == 0) {
            throw new IllegalStateException("address can not be empty!!!");
        }
        final List<Address> curAddresses = this.addresses.stream().filter(address -> address.healthy).collect(Collectors.toList());
        if (curAddresses.size() == 0) {
            return this.addresses.iterator().next().getAddress();
        }
        int index = count.getAndIncrement() % curAddresses.size();
        return curAddresses.get(index).getAddress();
    }

    public boolean needWait() {
        if (this.addresses == null || this.addresses.size() == 0) {
            throw new IllegalStateException("address can not be empty!!!");
        }
        final List<Address> curAddresses = this.addresses.stream().filter(address -> address.healthy).collect(Collectors.toList());
        return curAddresses.size() <= 1;
    }

    public void setOk(String okAddress) {
        for (Address address : addresses) {
            if (address.getAddress().equals(okAddress)) {
                address.setHealthy(true);
                address.setErrorTime(null);
                return;
            }
        }
    }

    public synchronized void setError(String errorAddress) {
        for (Address address : addresses) {
            if (address.getAddress().equals(errorAddress)) {
                address.setHealthy(false);
                address.setErrorTime(LocalDateTime.now());
                notify();
                return;
            }
        }
    }

    private class Address {

        final String address;

        volatile Boolean healthy;

        volatile LocalDateTime errorTime;

        public Address(String address, Boolean healthy, LocalDateTime errorTime) {
            this.address = address;
            this.healthy = healthy;
            this.errorTime = errorTime;
        }

        public void setHealthy(Boolean healthy) {
            this.healthy = healthy;
        }

        public void setErrorTime(LocalDateTime errorTime) {
            this.errorTime = errorTime;
        }

        public String getAddress() {
            return address;
        }

        public Boolean getHealthy() {
            return healthy;
        }

        public LocalDateTime getErrorTime() {
            return errorTime;
        }
    }

    private class AddressMaintenance implements Runnable {

        @Override
        public void run() {
            while (true) {
                if (allAddressIsOk()) {
                    waitForErrorHappen();
                }
                recoverIfTimeIsUp();
                sleep30s();
            }

        }

        private void waitForErrorHappen() {
            try {
                synchronized (AddressManager.this) {
                    AddressManager.this.wait();
                }
            } catch (InterruptedException e) {
                LOGGER.error("waitForErrorHappen interrupted!");
            }
        }

        private void sleep30s() {
            try {
                Thread.sleep(30000);
            } catch (InterruptedException e) {
                LOGGER.error("sleep30s interrupted!");
            }
        }

        private boolean allAddressIsOk() {
            return AddressManager.this.addresses.stream().allMatch(address -> address.healthy);
        }

        private void recoverIfTimeIsUp() {
            for (Address address : AddressManager.this.addresses) {
                LocalDateTime now = LocalDateTime.now();
                if (!address.healthy && address.errorTime != null && now.minusSeconds(failureRetryTime / 1000).isAfter(address.errorTime)) {
                    address.healthy = true;
                    address.errorTime = null;
                }
            }

        }

    }

}
